package com.itheime.test;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
 * 消息可靠性投递：
 * 1.确认模式 --》 producer 发送到 exchange  无论成功与否都会执行该方法
 * 2.回退模式 --》 当消息发送到exchange后，exchange路由到queue时失败时才会执行该方法
 */
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class Producer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 确认模式：
     * 执行的时间：
     * 生产者将生产的发送给exchange时，无论成功与否都会执行该方法，回调函数
     * 步骤：
     * 1.确实模式开启：ConnectionFactory中开启publisher-confirms="true"
     * 2.在rabbitTemplate定义ConfirmCallback回调函数
     */
    @Test
    public void testConfirm() {
        /**
         * correlationData: 相关配置信息
         * ack: exchange交换机是否成功收到了信息 true代表成功 false代表失败
         * cause: 失败原因
         */
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("confirm 回调方法被执行了");
                if (ack) {
                    System.out.println("exchange 接收信息成功！" + cause);
                } else {
                    //可根据回调判断信息是否发送成功，如果未发送成功可再次进行发送
                    System.out.println("exchange 未成功接受信息！" + cause);
                }
            }
        });
        //3.发送消息
        /**
         * 参数：
         * 1.交换机名称
         * 2.routingKey
         * 3.发送的消息
         */
        rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message");
    }

    /**
     * 回退模式：
     * 当消息发送给exchange后，exchange路由到queue失败时才会触发的该回调函数
     * 步骤：
     * 1.开启回退模式
     * 2.设置ReturnCallback
     * 3.设置Exchange处理消息的模式：
     * 1.如果消息没有路由到queue，则丢弃消息(默认)
     * 2.如果消息没有路由到queue，返回消息发送方
     */
    @Test
    public void testReturn() {
        //设置queue未接受消息的处理模式-->返回消息到发送方
        rabbitTemplate.setMandatory(true); // --> 设置 Exchange 处理消息的模式：
        //设置returnCallback
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             *
             * @param message 消息对象
             * @param i  错误码
             * @param s 错误信息
             * @param s1  交换机
             * @param s2  路由键
             */
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("return 被执行了 ");
                System.out.println(message);
                System.out.println(i);
                System.out.println(s);
                System.out.println(s1);
                System.out.println(s2);
            }
        });
        //3.发送消息
        /**
         * 参数：
         * 1.交换机名称
         * 2.routingKey
         * 3.发送的消息
         */
        rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm1", "message");
    }

    /**
     * 限流测试
     */
    @Test
    public void test() {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message");
        }
    }

    /**
     * ttl：过期时间
     * 1.队列同一过期
     * <p>
     * 2.消息单独过期
     * <p>
     * 当该消息处于队列顶端时，既即将被消费时会对其进行判断，是否过时，过时直接剔除！
     */
    @Test
    public void testTtl() {
        //设置消息过期时间 -->消息后处理对象，可以设置一些消息的参数信息
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //1.设置消息过期时长
                message.getMessageProperties().setExpiration("5000");
                return message;
            }
        };
        for (int i = 0; i < 10; i++) {
            if (i == 5) {
                rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.haha", "message...ttl", messagePostProcessor);
            } else {
                rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.haha", "message...ttl");
            }

        }
    }

    /**
     * 延迟队列数据生成
     */
    @Test
    public void testDlx() throws InterruptedException {
        rabbitTemplate.convertAndSend("order_exchange", "order.msg", "订单信息：id*1,华为mate20");
        for (int i = 10; i > 0 ; i--) {
            System.out.println(i+"*********");
            Thread.sleep(1000);
        }
    }
}
